package com.alisonyu.spider.Scheduler;

import com.alisonyu.spider.Downloader.Request;
import com.alisonyu.spider.SpiderConfig;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;

public class RedisScheduler implements Schedule{

    private JedisPool redisPool;
    private String queueKey;
    private String setKey;
    private Logger logger = LoggerFactory.getLogger(RedisScheduler.class.getName());

    public RedisScheduler(){}

    public RedisScheduler(JedisPool pool, SpiderConfig config){
        this.redisPool = pool;
        this.queueKey = config.getRedisUrlQueueKey();
        this.setKey = config.getRedisUrlSetKey();

    }

    @Override
    public Request getRequest() {
        try(Jedis redis = redisPool.getResource()){
            String requestJson = redis.lpop(queueKey);
            if (requestJson==null) return null;
            logger.info("从队列中获取到:"+requestJson);
            return Request.toRequest(new JsonObject(requestJson));
        }
    }

    @Override
    public void putUrl(String url) {
       putRequest(new Request(url));
    }

    @Override
    public void putRequest(Request request) {
        try (Jedis redis = redisPool.getResource()) {
            try(Pipeline pipeline = redis.pipelined()){
                pipeline.watch(setKey);
                Response<Boolean> exist = pipeline.sismember(setKey, request.getUrl());
                pipeline.sync();
                if (!exist.get()) {
                    pipeline.multi();
                    pipeline.rpush(queueKey, request.toJsonObject().toString());
                    pipeline.sadd(setKey, request.getUrl());
                    pipeline.exec();
                    logger.info(request.getUrl()+"添加到爬虫队列");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
